package cn.fintecher.pangolin.service.report.scheduled;

import cn.fintecher.pangolin.common.utils.Constants;
import cn.fintecher.pangolin.common.utils.ZWDateUtil;
import cn.fintecher.pangolin.entity.elastic.DailyCollectionStatisticsRecord;
import cn.fintecher.pangolin.service.report.elastic.DailyCollectionStatisticsRecordRepository;
import cn.fintecher.pangolin.service.report.model.SettleCaseModel;
import cn.fintecher.pangolin.service.report.repository.CaseInfoRepository;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.valuecount.InternalValueCount;
import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCountAggregationBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigInteger;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author wangxiangdong
 * @Description 每日催收统计
 * @Date 2019/12/10 14:48
 **/
@Component
@Slf4j
@EnableScheduling
public class DailyCollectionStatisticsScheduled {

    @Autowired
    private CaseInfoRepository caseInfoRepository;

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    private DailyCollectionStatisticsRecordRepository dailyCollectionStatisticsRecordRepository;

    @Scheduled(cron = "0 40 23 * * ?")
    public void dailyCollectionStatisticsScheduled(){
        log.info("每日催收统计开始:{}", ZWDateUtil.getFormatNowDateTime("yyyy-MM-dd HH:mm:ss"));
        List<DailyCollectionStatisticsRecord> dailyCollectionStatisticsRecordList =  new ArrayList<>();
        List<SettleCaseModel> settleCaseModelList = new ArrayList<>();
        List<Object[]> dataRecords = caseInfoRepository.collectDataStatistics();
        if(dataRecords.size()>0){
            excuteDataRecord(dailyCollectionStatisticsRecordList,dataRecords);
        }else{
            log.info("没有在催的案件:{}", ZWDateUtil.getFormatNowDateTime("yyyy-MM-dd HH:mm:ss"));
            return;
        }
        Date currentDate =  ZWDateUtil.getNowDate();
        Map<String,DailyCollectionStatisticsRecord> dailyCollectionStatisticsRecordMap = dailyCollectionStatisticsRecordList.stream().collect(Collectors.toMap(DailyCollectionStatisticsRecord::getCollectorId,dailyCollectionStatisticsRecord -> dailyCollectionStatisticsRecord));
        //查询今日结清案件caseId
        List<Object[]> settleData = caseInfoRepository.querySettleCaseIds(dailyCollectionStatisticsRecordMap.keySet(), currentDate);
        if(!settleData.isEmpty()){
            settleData.forEach(settleDataArr->{
                SettleCaseModel settleCaseModel = new SettleCaseModel();
                settleCaseModel.setCaseId((String) settleDataArr[0]);
                settleCaseModel.setCollectorId((String)settleDataArr[1]);
                settleCaseModelList.add(settleCaseModel);
            });
        }
        //查询对账单（结清金额）
        Map<String,List<SettleCaseModel>> stringListMap = settleCaseModelList.stream().collect(Collectors.groupingBy(SettleCaseModel::getCollectorId));
        for(Map.Entry<String,List<SettleCaseModel>>  stringListEntry: stringListMap.entrySet()){
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            boolQueryBuilder.must(QueryBuilders.termsQuery("caseId.keyword",stringListEntry.getValue().stream().map(SettleCaseModel::getCaseId).collect(Collectors.toList())));
            boolQueryBuilder.must(QueryBuilders.termQuery("flag",0));
            SumAggregationBuilder sumAggregationBuilder = AggregationBuilders.sum("paymentTotal").field("paymentAmt");
            SearchQuery searchQuery = new NativeSearchQueryBuilder().withIndices(Constants.PAYMENT_RECORD)
                    .withTypes(Constants.PAYMENT_RECORD).withSearchType(SearchType.DEFAULT).withQuery(boolQueryBuilder).addAggregation(sumAggregationBuilder).build();
            Aggregations aggregations = elasticsearchTemplate.query(searchQuery, SearchResponse::getAggregations);
            Map<String, Aggregation> aggregationMap = aggregations.getAsMap();
            double paymentTotal = ((InternalSum) aggregationMap.get("paymentTotal")).getValue();
            DailyCollectionStatisticsRecord dailyCollectionStatisticsRecord = dailyCollectionStatisticsRecordMap.get(stringListEntry.getKey());
            dailyCollectionStatisticsRecord.setSettltAmt(paymentTotal);
            dailyCollectionStatisticsRecord.setSettltCount(stringListEntry.getValue().size());
        }
        //统计流入流出数量、金额
        for(Map.Entry<String,DailyCollectionStatisticsRecord> dailyCollectionStatisticsRecordEntry: dailyCollectionStatisticsRecordMap.entrySet()){
            //当天流出该催收员的案件ID
            List<String> outFlowList = new ArrayList<>();
            BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
            queryBuilder.must(QueryBuilders.termQuery("outUserId.keyword",dailyCollectionStatisticsRecordEntry.getKey()));
            queryBuilder.must(QueryBuilders.rangeQuery("operatorTime").gt(currentDate.getTime()).lt(currentDate.getTime() + 86400000));
            ValueCountAggregationBuilder valueCountAggregator = AggregationBuilders.count("outflowCount").field("outUserId.keyword");
            SumAggregationBuilder sumAggregationBuilder = AggregationBuilders.sum("leftAmtTotal").field("leftAmt");
            TermsAggregationBuilder outFlowTermsAggregation = AggregationBuilders.terms("caseIds").field("caseId.keyword").size(100000);
            SearchQuery searchQuery = new NativeSearchQueryBuilder().withIndices(Constants.CASE_TRANSFER_LOG)
                    .withTypes(Constants.CASE_TRANSFER_LOG).withSearchType(SearchType.DEFAULT).withQuery(queryBuilder)
                    .addAggregation(outFlowTermsAggregation).addAggregation(valueCountAggregator).addAggregation(sumAggregationBuilder).build();
            Aggregations aggregations = elasticsearchTemplate.query(searchQuery, SearchResponse::getAggregations);
            Map<String, Aggregation> aggregationMap = aggregations.getAsMap();
            DailyCollectionStatisticsRecord dailyCollectionStatisticsRecord = dailyCollectionStatisticsRecordMap.get(dailyCollectionStatisticsRecordEntry.getKey());
            long caseCount = ((InternalValueCount) aggregationMap.get("outflowCount")).getValue();
            double outFlowAmount = ((InternalSum) aggregationMap.get("leftAmtTotal")).getValue();
            StringTerms outFlowStringTerms = (StringTerms) aggregationMap.get("caseIds");
            for (StringTerms.Bucket bucket : outFlowStringTerms.getBuckets()) {
                outFlowList.add(bucket.getKeyAsString());
            }
            dailyCollectionStatisticsRecord.setOutflowAmount(outFlowAmount);
            dailyCollectionStatisticsRecord.setOutflowCount(caseCount);
            //当天流入到该催收员的案件ID
            List<String> inFlowList = new ArrayList<>();
            //当前催收员持有的案件
            List<String> caseIds = caseInfoRepository.queryCaseIds(dailyCollectionStatisticsRecordEntry.getKey());
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            boolQueryBuilder.must(QueryBuilders.termQuery("inUserId.keyword",dailyCollectionStatisticsRecordEntry.getKey()));
            boolQueryBuilder.must(QueryBuilders.rangeQuery("operatorTime").gte(currentDate.getTime()).lt(currentDate.getTime() + 86400000));
            TermsAggregationBuilder inFlowTermsAggregation = AggregationBuilders.terms("caseIds").field("caseId.keyword").size(100000);
            ValueCountAggregationBuilder inflowCountAggregations = AggregationBuilders.count("inflowCount").field("inUserId.keyword");
            SumAggregationBuilder sumAggregation = AggregationBuilders.sum("leftAmtTotal").field("leftAmt");
            SearchQuery inFlowSearchQuery = new NativeSearchQueryBuilder().withIndices(Constants.CASE_TRANSFER_LOG)
                    .withTypes(Constants.CASE_TRANSFER_LOG).withQuery(boolQueryBuilder)
                    .addAggregation(inFlowTermsAggregation).addAggregation(inflowCountAggregations).addAggregation(sumAggregation).build();
            Aggregations aggregations2 = elasticsearchTemplate.query(inFlowSearchQuery, SearchResponse::getAggregations);
            Map<String, Aggregation> inFlowAggregationMap = aggregations2.getAsMap();
            long inflowCount = ((InternalValueCount) inFlowAggregationMap.get("inflowCount")).getValue();
            StringTerms inFlowStringTerms = (StringTerms) inFlowAggregationMap.get("caseIds");
            for (StringTerms.Bucket bucket : inFlowStringTerms.getBuckets()) {
                inFlowList.add(bucket.getKeyAsString());
            }
            List<String> diffrentCaseIds = getDiffrentCaseIds(caseIds, inFlowList);
            List<String> inOutDiff = getDiffrentCaseIds(inFlowList,outFlowList);
            if(diffrentCaseIds.size() > 0){
                double leftAmountTotal = caseInfoRepository.queryLeftAmountTotal(diffrentCaseIds);
                dailyCollectionStatisticsRecord.setCollectAmt(dailyCollectionStatisticsRecord.getCollectAmt() + leftAmountTotal);
            }
            if(inOutDiff.size() > 0){
                double leftAmountTotal = caseInfoRepository.queryLeftAmountTotal(inOutDiff);
                dailyCollectionStatisticsRecord.setCollectAmt(dailyCollectionStatisticsRecord.getCollectAmt() + leftAmountTotal);
            }

            dailyCollectionStatisticsRecord.setCollectCount(inOutDiff.size() + diffrentCaseIds.size() + dailyCollectionStatisticsRecord.getCollectCount());
            double inFlowAmount = ((InternalSum) inFlowAggregationMap.get("leftAmtTotal")).getValue();
            dailyCollectionStatisticsRecord.setInflowAmount(inFlowAmount);
            dailyCollectionStatisticsRecord.setInflowCount(inflowCount);
        }
        List<DailyCollectionStatisticsRecord> dailyCollectionStatisticsRecords = new ArrayList<>(dailyCollectionStatisticsRecordMap.values());
        dailyCollectionStatisticsRecordRepository.saveAll(dailyCollectionStatisticsRecords);
        log.info("每日催收统计结束:{}", ZWDateUtil.getFormatNowDateTime("yyyy-MM-dd HH:mm:ss"));
    }


    /**
     * @Author wangxiangdong
     * @Description 转换统计数据
     * @Date 2019/12/16 16:39
     **/
    private void excuteDataRecord(List<DailyCollectionStatisticsRecord> dailyCollectionStatisticsRecordList, List<Object[]> dataRecords){
        dataRecords.forEach(dataRecord->{
            DailyCollectionStatisticsRecord dailyCollectionStatisticsRecord = new DailyCollectionStatisticsRecord();
            dailyCollectionStatisticsRecord.setId(UUID.randomUUID().toString().replaceAll("/",""));
            dailyCollectionStatisticsRecord.setDepartmentId((String) dataRecord[0]);
            dailyCollectionStatisticsRecord.setDepartmentName((String)dataRecord[1]);
            dailyCollectionStatisticsRecord.setDepartmentCode((String)dataRecord[2]);
            dailyCollectionStatisticsRecord.setCollectorId((String)dataRecord[3]);
            dailyCollectionStatisticsRecord.setCollectorName((String)dataRecord[4]);
            dailyCollectionStatisticsRecord.setCollectCount(((BigInteger)dataRecord[5]).intValue());
            dailyCollectionStatisticsRecord.setCollectAmt((double)dataRecord[6]);
            dailyCollectionStatisticsRecord.setStatisticsDate(ZWDateUtil.getNowDate());
            dailyCollectionStatisticsRecordList.add(dailyCollectionStatisticsRecord);
        });
    }

    /**
     * @Author wangxiangdong
     * @Description  抓取第二个集合与第一个集合不同的元素
     * @Date 2019/12/27 9:22
     **/
    private static List<String> getDiffrentCaseIds(List<String> list1, List<String> list2) {
        List<String> idsList1 = new ArrayList<>(list1);
        List<String> idsList2 = new ArrayList<>(list2);
        if(idsList1.size()>0){
            idsList2.removeAll(idsList1);
            return idsList2;
        }else{
            return idsList2;
        }
    }


}
